package cl.uchile.cos.sync;

import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;

import cl.uchile.cos.persistance.StateManager;
import cl.uchile.cos.sync.adapters.ServerAdapter;
import cl.uchile.cos.sync.listeners.CouplingListener;
import cl.uchile.cos.sync.listeners.Listener;
import cl.uchile.cos.sync.listeners.MessageListener;

public class CouplingManagerImpl implements CouplingManager {
	/**
	 * 
	 */
	private static final long serialVersionUID = 6184655961711874295L;

	final class NextInQueue implements Runnable {
		private CouplingManager manager;

		public NextInQueue(CouplingManager manager) {
			this.manager = manager;
		}

		@Override
		public void run() {
			synchronized (this.manager) {
				try {
					QueuedTask task = queue.poll();
					if (task == null)
						executor.execute(new NextInQueue(this.manager));
					else {
						task.execute();
					}
				} catch (Exception e) {
					e.printStackTrace();
					executor.execute(new NextInQueue(this.manager));
				}

			}
		}
	}

	final class CouplerTask implements QueuedTask {
		private String objectId;
		private ServerAdapter adapter;
		private List<Message> messages;
		private Long timestamp;

		public CouplerTask(String objectId, ServerAdapter adapter,
				List<Message> messages, Long timestamp) {
			super();
			this.objectId = objectId;
			this.adapter = adapter;
			this.messages = messages;
			this.timestamp = timestamp;
		}

		@Override
		public void execute() {
			doCouple(this.objectId, this.adapter, this.messages, this.timestamp);

		}

	}

	private Map<String, Set<ServerAdapter>> adapters;
	private Set<MessageListener> messageListeners;
	private Set<CouplingListener> couplingListener;
	private StateManager stateManager;
	private Long messagesToGo = 0L;
	private transient Executor executor;
	private String sessionId;
	private Long lastUsedTimestamp = 0L;
	private Map<String, Long> couplingTimestamp = new HashMap<String, Long>();
	private Queue<QueuedTask> queue = new ConcurrentLinkedQueue<QueuedTask>();

	public Set<MessageListener> getMessageListeners() {
		return messageListeners;
	}

	public Set<CouplingListener> getCouplingListener() {
		return couplingListener;
	}

	public String getSessionId() {
		return sessionId;
	}

	public void setSessionId(String sessionId) {
		this.sessionId = sessionId;
		this.stateManager.setSessionId(sessionId);
	}

	public CouplingManagerImpl(Executor executor) {
		this.adapters = new LinkedHashMap<String, Set<ServerAdapter>>();
		this.messageListeners = new LinkedHashSet<MessageListener>();
		this.couplingListener = new LinkedHashSet<CouplingListener>();
		this.executor = executor;
	}

	@Override
	public synchronized void addListener(Listener listener) {
		if (MessageListener.class.isInstance(listener))
			this.messageListeners.add((MessageListener) listener);
		else if (CouplingListener.class.isInstance(listener)) {
			this.couplingListener.add((CouplingListener) listener);
		}
	}

	@Override
	public synchronized void removeListener(Listener listener) {
		if (MessageListener.class.isInstance(listener))
			this.messageListeners.remove((MessageListener) listener);
		else if (CouplingListener.class.isInstance(listener)) {
			this.couplingListener.remove((CouplingListener) listener);
		}
	}

	public synchronized void propagate(Message message, ServerAdapter adapter,
			boolean echo) {

		String id = message.getObjectId();
		messagesToGo--;
		if (this.stateManager != null)
			this.stateManager.processMessage(message);
		for (ServerAdapter destAdapter : adapters.get(id)) {
			if ((!echo && destAdapter.equals(adapter))
			/* || isFromTheFuture(destAdapter.getClient().getId(), message) */)
				continue;
			destAdapter.sendToClient(message);
			for (MessageListener listener : this.messageListeners) {
				listener.onSendToAdapter(message, adapter);
			}
		}
		for (MessageListener listener : this.messageListeners) {
			listener.onPropagationEnd(message);
		}
	}

	/**
	 * Regista el mensaje para ser propadgado. Si el mensaje no es seguro por
	 * temas de sincronizacion retornara false y lo reenviara sincronizado al
	 * cliente en el futuro. De lo contrario no se lo enviara denuevo al cliente
	 * que lo envio.
	 * 
	 * @param message
	 * @param adapter
	 * @return true si el cliente puede ejecutar el evento, false si no.
	 */
	@Override
	public synchronized boolean registerMessage(Message message,
			ServerAdapter adapter) {
		return this.registerMessage(message, adapter, false);

	}

	@Override
	public synchronized boolean registerMessage(Message message,
			ServerAdapter adapter, boolean echo) {
		boolean safe = false;
		try {
			if (this.stateManager != null
					&& !this.stateManager.validateMessage(message, adapter)) {
				return false;
			}

			message.setTimestamp(System.currentTimeMillis());

			// This prevent 2 messages to have the same timestamp at the small
			// (if any) cost.
			while (message.getTimestamp() == lastUsedTimestamp) {
				Thread.sleep(1);
				message.setTimestamp(System.currentTimeMillis());
			}
			lastUsedTimestamp = message.getTimestamp();
			safe = isMessageSyncSafe(message, adapter);
			this.messagesToGo++;
			this.queue.offer(new MessagePropagator(message, adapter, !safe
					|| echo, this));
			this.executor.execute(new NextInQueue(this));
		} catch (InterruptedException e) {
			// TODO logging
			e.printStackTrace();
		}
		return safe;
	}

	/**
	 * Revisa si el mensaje puede o no causar problemas de sincronizacion con el
	 * cliente. Esto significa revisar si hay algun mensaje emitido antes que
	 * aun no ha sido entregado a al cliente que envio este mensaje.
	 * 
	 * @param message
	 * @return
	 */
	private synchronized boolean isMessageSyncSafe(Message message,
			ServerAdapter adapter) {
		if (adapter.ready()) {
			if (this.messagesToGo == 0)
				return true;
		}
		return false;
	}

	@Override
	public synchronized void couple(String objectId, ServerAdapter adapter) {
		this.couple(objectId, adapter, null);
	}

	@Override
	public synchronized void decouple(String objectId, ServerAdapter adapter) {
		if (this.adapters.containsKey(objectId)) {
			this.adapters.get(objectId).remove(adapter);
		}
		for (CouplingListener listener : this.couplingListener) {
			listener.onDecouple(objectId, adapter);
		}
	}

	@Override
	public synchronized void decouple(ServerAdapter adapter) {
		for (String objectId : this.adapters.keySet()) {
			this.decouple(objectId, adapter);
		}
	}

	@Override
	public void setStateManager(StateManager stateManager) {
		this.stateManager = stateManager;
	}

	@Override
	public synchronized void couple(String objectId, ServerAdapter adapter,
			List<Message> messages) {
		long timestamp = System.currentTimeMillis();
		while (timestamp <= lastUsedTimestamp) {
			try {
				Thread.sleep(1);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			timestamp = System.currentTimeMillis();
		}
		lastUsedTimestamp = timestamp;
		this.queue
				.offer(new CouplerTask(objectId, adapter, messages, timestamp));
		this.executor.execute(new NextInQueue(this));
	}

	private synchronized void doCouple(String objectId, ServerAdapter adapter,
			List<Message> messages, Long timestamp) {
		if (this.stateManager != null) {
			adapter.initializeObjectData(objectId, stateManager, timestamp);
		}
		// this.confirmCouple(objectId, adapter.getClient().getId(), timestamp);
		if (!this.adapters.containsKey(objectId)) {
			this.adapters.put(objectId, new LinkedHashSet<ServerAdapter>());
		}
		this.adapters.get(objectId).add(adapter);
		for (CouplingListener listener : this.couplingListener) {
			listener.onCouple(objectId, adapter);
		}
		if (stateManager.getPersistedStateMessageCount(objectId) == 0
				&& messages != null) {
			for (Message message : messages) {
				this.registerMessage(message, adapter, true);
			}
		}
	}

	/*
	 * private boolean isFromTheFuture(Long clientId, Message m) { Long
	 * timestamp = this.couplingTimestamp.get(m.getObjectId() + "-" + clientId);
	 * boolean delFuturo = timestamp > m.getTimestamp(); if (delFuturo)
	 * System.out.println(String.format(
	 * "A LA Mierda!!! Cliente %s (%s) es del futuro %s", clientId,
	 * timestamp.toString(), m.getTimestamp() + "")); return delFuturo; }
	 * 
	 * private void confirmCouple(String objectId, Long clientId, Long
	 * timestamp) { System.out.println(String.format(
	 * "Cliente %s se acoplado con timestamp %s", clientId,
	 * timestamp.toString())); this.couplingTimestamp.put(objectId + "-" +
	 * clientId, timestamp); }
	 */
	@Override
	public void destroy() {
		if (this.stateManager != null)
			this.stateManager.destroy();

	}

	static public void main(String[] args) {
		for (int i = 0; i < 100; ++i) {
			System.out.println(System.currentTimeMillis());
		}
	}
}
